package com.xujian.cdc.schema;

import com.alibaba.fastjson2.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.ververica.cdc.debezium.utils.TemporalConversions;
import com.xujian.common.beans.JsonRecord;
import io.debezium.data.Envelope;
import io.debezium.time.*;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.flink.util.StringUtils;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.TimeZone;

/**
 * @author star xu
 * @date 2023/1/3 15:25
 * @Description: TODO 日期时区还有问题，全量和增量都差8个小时
 * @Slogan: 致敬大师,致敬未来的自己
 */
public class JsonDebeziumDeserializationSchema
        implements DebeziumDeserializationSchema<JsonRecord> {

    private static final String TIME_ZONE_ID = "Asia/Shanghai";


    @Override
    public void deserialize(SourceRecord record, Collector<JsonRecord> out) {
        Envelope.Operation op = Envelope.operationFor(record);
        Struct value = (Struct) record.value();
        Schema valueSchema = record.valueSchema();
        String tableName = record.topic();
        //out.collect("source table name is :" + tableName);
        if (op == Envelope.Operation.CREATE || op == Envelope.Operation.READ) {
            JSONObject insertMapString = extractAfterRow(value, valueSchema);
            JsonRecord jsonRecord = new JsonRecord(tableName, "i", insertMapString.toJSONString());
            out.collect(jsonRecord);
        } else if (op == Envelope.Operation.DELETE) {
            JSONObject deleteString = extractBeforeRow(value, valueSchema);
            // starrocks表需要使用主键模型，另外json中需要有{"__op":1}表示删除,{"__op":0}表示upsert
            deleteString.put("__op", 1);
            JsonRecord jsonRecord = new JsonRecord(tableName, "d", deleteString.toJSONString());
            out.collect(jsonRecord);
        } else if (op == Envelope.Operation.UPDATE) {
            JSONObject updateString = extractAfterRow(value, valueSchema);
            // starrocks表需要使用主键模型，另外json中需要有{"__op":1}表示删除,{"__op":0}表示upsert
            updateString.put("__op", 0);
            JsonRecord jsonRecord = new JsonRecord(tableName, "u", updateString.toJSONString());
            out.collect(jsonRecord);
        }
    }

    @Override
    public TypeInformation<JsonRecord> getProducedType() {
        return TypeInformation.of(new TypeHint<JsonRecord>() {
        });
    }


    private JSONObject extractAfterRow(Struct value, Schema valueSchema) {
        Struct after = value.getStruct(Envelope.FieldName.AFTER);
        Schema afterSchema = valueSchema.field(Envelope.FieldName.AFTER).schema();
        return transDataFun(after);
    }

    private JSONObject extractBeforeRow(Struct value, Schema valueSchema) {
        Struct beforeValue = value.getStruct(Envelope.FieldName.BEFORE);
        Schema beforeSchema = valueSchema.field(Envelope.FieldName.BEFORE).schema();
        return transDataFun(beforeValue);
    }

    private Map<String, Object> getRowMap(Struct value, Schema valueSchema) {
        Map<String, Object> map = new HashMap<>();
        String TIME_ZONE_ID = "Asia/Shanghai";
        for (Field field : valueSchema.fields()) {
            if ("int64".equals(field.schema().type().getName()) && "io.debezium.time.Timestamp".equals(field.schema().name()) && value.get(field) != null) {
                TimeZone timeZone = StringUtils.isNullOrWhitespaceOnly(TIME_ZONE_ID) ? TimeZone.getDefault() : TimeZone.getTimeZone(ZoneId.of(TIME_ZONE_ID));
                Date date = new Date((Long) value.get(field) - timeZone.getRawOffset());
                map.put(field.name(), date);
            } else {
                map.put(field.name(), value.get(field.name()));
            }
        }
        return map;
    }

    public JSONObject transData(Struct struct){
        JSONObject result = new JSONObject();
        String TIME_ZONE_ID = "Asia/Shanghai";
        if (struct != null){
            for (Field field : struct.schema().fields()) {
                if ("int64".equals(field.schema().type().getName()) && "io.debezium.time.Timestamp".equals(field.schema().name()) && struct.get(field) != null) {
                    TimeZone timeZone = StringUtils.isNullOrWhitespaceOnly(TIME_ZONE_ID) ? TimeZone.getDefault() : TimeZone.getTimeZone(ZoneId.of(TIME_ZONE_ID));
                    Date date = new Date((Long) struct.get(field) - timeZone.getRawOffset());
                    result.put(field.name(), date);
                } else {
                    result.put(field.name(), struct.get(field));
                }
            }
        }
        return result;
    }

    public JSONObject transDataFun(Struct struct){
        JSONObject result = new JSONObject();
        if (struct != null){
            for (Field field : struct.schema().fields()) {
                Object obj = struct.get(field);
                if (obj == null){
                    result.put(field.name(), null);
                }else if (obj instanceof Long){
                    obj = convertLongToTime((Long) struct.get(field), field.schema().name());
                    result.put(field.name(), obj);
                }else if (obj instanceof Integer){
                    obj = convertIntToDate((Integer) struct.get(field),field.schema().name());
                    result.put(field.name(), obj);
                }else{
                    obj = convertObjToTime(struct.get(field), field.schema().name());
                    result.put(field.name(), obj);
                }
            }
        }
        return result;
    }

    private Object convertLongToTime(Long obj,String type){
        if (org.apache.commons.lang3.StringUtils.isBlank(type)){
            return obj;
        }
        SchemaBuilder time_schema = SchemaBuilder.int64().name("org.apache.kafka.connect.data.Time");
        SchemaBuilder timestamp_schema = SchemaBuilder.int64().name("org.apache.kafka.connect.data.Timestamp");
        SchemaBuilder date_schema = SchemaBuilder.int64().name("org.apache.kafka.connect.data.Date");
        Object result = obj;
        switch (type){
            case Time.SCHEMA_NAME:
                result =  org.apache.kafka.connect.data.Time.toLogical(time_schema, obj.intValue()).toInstant().atZone(ZoneId.of(TIME_ZONE_ID)).toLocalTime().toString();
                break;
            case MicroTime.SCHEMA_NAME:
                result =  org.apache.kafka.connect.data.Time.toLogical(time_schema, (int) (obj / 1000)).toInstant().atZone(ZoneId.of(TIME_ZONE_ID)).toLocalTime().toString();
                break;
            case NanoTime.SCHEMA_NAME:
                result =  org.apache.kafka.connect.data.Time.toLogical(time_schema, (int) (obj / 1000 / 1000)).toInstant().atZone(ZoneId.of(TIME_ZONE_ID)).toLocalTime().toString();
                break;
            case Timestamp.SCHEMA_NAME:
                LocalDateTime localDateTime = org.apache.kafka.connect.data.Timestamp.toLogical(timestamp_schema, obj).toInstant().atZone(ZoneId.of(TIME_ZONE_ID)).toLocalDateTime();
                result = java.sql.Timestamp.valueOf(localDateTime).toString();
                break;
            case MicroTimestamp.SCHEMA_NAME:
                LocalDateTime localDateTime1 =  org.apache.kafka.connect.data.Timestamp.toLogical(timestamp_schema, obj / 1000).toInstant().atZone(ZoneId.of(TIME_ZONE_ID)).toLocalDateTime();
                result = java.sql.Timestamp.valueOf(localDateTime1).toString();
                break;
            case NanoTimestamp.SCHEMA_NAME:
                LocalDateTime localDateTime2 =  org.apache.kafka.connect.data.Timestamp.toLogical(timestamp_schema, obj / 1000 / 1000).toInstant().atZone(ZoneId.of(TIME_ZONE_ID)).toLocalDateTime();
                result = java.sql.Timestamp.valueOf(localDateTime2).toString();
                break;
            case io.debezium.time.Date.SCHEMA_NAME:
                result =  org.apache.kafka.connect.data.Date.toLogical(date_schema, obj.intValue()).toInstant().atZone(ZoneId.of(TIME_ZONE_ID)).toLocalDate().toString();
                break;
            default:
        }
        return result;
    }

    private Object convertIntToDate(int obj, String type){
        if (org.apache.commons.lang3.StringUtils.isBlank(type)){
            return obj;
        }
        SchemaBuilder date_schema = SchemaBuilder.int64().name("org.apache.kafka.connect.data.Date");
        Object result = obj;
        if (io.debezium.time.Date.SCHEMA_NAME.equals(type)) {
            result = org.apache.kafka.connect.data.Date.toLogical(date_schema, obj).toInstant().atZone(ZoneId.of(TIME_ZONE_ID)).toLocalDate().toString();
        }
        return result;
    }

    private Object convertObjToTime(Object obj, String type){
        if (org.apache.commons.lang3.StringUtils.isBlank(type)){
            return obj;
        }
        Object result = obj;
        switch (type){
            case Time.SCHEMA_NAME:
            case  MicroTime.SCHEMA_NAME:
            case NanoTime.SCHEMA_NAME:
                result = java.sql.Time.valueOf(TemporalConversions.toLocalTime(obj)).toString();
                break;
            case Timestamp.SCHEMA_NAME:
            case  MicroTimestamp.SCHEMA_NAME:
            case  NanoTimestamp.SCHEMA_NAME:
            case ZonedTimestamp.SCHEMA_NAME:
                result = java.sql.Timestamp.valueOf(TemporalConversions.toLocalDateTime(obj, ZoneId.of(TIME_ZONE_ID))).toString();
                break;
            default:
        }
        return result;
    }
}